package io.renren.modules.rabbitmq.listener;

import cn.hutool.core.util.RandomUtil;
import com.alibaba.fastjson.JSONObject;
import com.rabbitmq.client.Channel;
import io.renren.common.utils.RedisUtils;
import io.renren.modules.gasstation.realtimedata.entity.RealTimeDataEntity;
import io.renren.modules.gasstation.realtimedata.service.RealTimeDataService;
import io.renren.modules.rabbitmq.config.MqConfig;
import io.renren.modules.rabbitmq.constant.Constant;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.io.IOException;

/**
 * @Author：T-weiwei
 * @Description: 加油站业务监听类
 * @Email：tweiwei@aliyun.com
 * @CreateDate：2022-01-27-上午10:16
 * @CopyRight：http://www.huaxun.ink/
 * @Hint： This content is limited to the internal circulation of HuaXun Hi Tech Co., Ltd.,
 * and is forbidden to be leaked and used for other commercial purposes
 **/
@Component(value = "fanout-receiver")
public class GasStationBusinessListener {

    /**
     * redis工具类
     */
    @Autowired
    private RedisUtils redisUtils;

    /**
     * 加油站实时数据业务层
     */
    @Autowired
    private RealTimeDataService realTimeDataService;

    /**
     * 监听大气队列中的消息并消费
     * @param content 传来的内容
     * @param channel 管道
     * @param message 消息体
     * @throws IOException 确认该消息所抛出的异常
     */
    @RabbitHandler
    @RabbitListener(queues = {MqConfig.GAS_STATION_QUEUE})
    public void listeningToTheGasStationQueueMessage(String content, Channel channel, Message message) throws IOException {
        try {
            // TODO 具体业务逻辑(注意: 消费者如果要接收的是实体类的话,需要对实体类进行序列化.并且将形式参数: content的类型跟换为相应实体类即可)
            JSONObject json = JSONObject.parseObject(content);
            JSONObject data_array = JSONObject.parseObject(json.get("data_array").toString());
            RealTimeDataEntity entity = JSONObject.toJavaObject(data_array, RealTimeDataEntity.class);
            entity.setObjectId(RandomUtil.randomString(8));
            this.realTimeDataService.save(entity);
            // 手动确认该消息,第二个参数翻译过来是多种多样,倍数,数量多的.所以第二个参数代表是否为多消息的意思.
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        }  catch (Exception e) {
            // 通过key获取它已经重复提交的次数
            String key = message.getMessageProperties().getHeaders().get(Constant.SPRING_RETURNED_MESSAGE_CORRELATION).toString();
            // 如果重复回滚到队列次数大于等于一次
            if (this.redisUtils.hasKey(key)) {
                Integer retryCount = Integer.valueOf(this.redisUtils.get(key));

                if (retryCount >= Constant.MAX_RETRIES) {
                    // 超出重试次数,放到死信队列中
                    channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
                } else {
                    // 再次放入到队列中,并在redis中再次增加次数
                    channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
                    this.redisUtils.set(key, (retryCount + 1) + "");
                }
            } else {
                // 初次回滚到队列中,并在redis中做记录.
                channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
                this.redisUtils.set(key, Constant.INITIALIZATION_RETRIES);
            }
        }
    }
}
